package org.itstack.demo.netty.server;

import com.sun.xml.internal.messaging.saaj.util.ByteInputStream;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.stream.ChunkedStream;
import io.netty.util.ReferenceCountUtil;

import java.nio.ByteBuffer;

/**
 * 流量分块实现
 */
public class MyServerChunkHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        // 内容验证
        if (!(msg instanceof ByteBuf)) {
            super.write(ctx, msg, promise);
            return;
        }
        // 获取byte
        ByteBuf byteBuf = (ByteBuf) msg;
        byte[] data = this.getData(byteBuf);

        // 写入流中
        ByteInputStream in = new ByteInputStream();
        in.setBuf(data);

        // 消息分块，十个字节，测试过程中可以调整
        ChunkedStream chunkedStream = new ChunkedStream(in, 10);
        // 管道消息传输承诺
        ChannelProgressivePromise progressivePromise = ctx.channel().newProgressivePromise();
        progressivePromise.addListener(new ChannelProgressiveFutureListener() {
            @Override
            public void operationProgressed(ChannelProgressiveFuture channelProgressiveFuture, long l, long l1) throws Exception {

            }

            @Override
            public void operationComplete(ChannelProgressiveFuture future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println("消息发送成功 success");
                    promise.setSuccess();
                } else {
                    System.out.println("消息发送失败 failure");
                    promise.setFailure(future.cause());
                }
            }
        });

        ReferenceCountUtil.release(msg);
        ctx.write(chunkedStream, progressivePromise);

    }

    // 获取Byte
    public byte[] getData(ByteBuf byteBuf) {
        if (byteBuf.hasArray()) {
            return byteBuf.array().clone();
        }
        byte[] data = new byte[byteBuf.readableBytes() - 1];
        byteBuf.readBytes(data);
        return data;
    }


}
